package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 使用这个类来统一管理内存中的数据
 */
public class MemoryDataCenter {

    // key是name value是Exchange对象
    //为了保证线程安全 我们使用线程安全的容器
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    //key是name value是MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();

    //使用一个嵌套的 HashMap key是exchangeName  value又是一个HashMap  key是QueueName value 是Binding对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();

    //key是messageId  value是Message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();

    // key是queueName value又是一个LinkedList 每个元素是一个Message对象
    // 表示队列和消息的关联关系
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();

    //key是queueName value又是一个 ConcurrentHashMap  key是消息id value 是 Message对象
    //表示未被确认的关系
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();


    //根据name 获取交换机对象
    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    //插入交换机
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功 exchangeName="+exchange.getName());
    }

    //删除交换机
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功 exchangeName="+exchangeName);
    }

    //获取队列
    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    //插入队列
    public void insertQueue(MSGQueue queue){
        queueMap.putIfAbsent(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 队列添加成功 queueName="+queue.getName());
    }

    //删除队列
    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功 queueName="+queueName);
    }

    //根据exchangeName和queueName 获取唯一绑定对象
    public Binding getBinding(String exchangeName,String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap==null){
            //没有查询到 说明没有绑定关系
            return null;
        }
        return bindingMap.get(queueName);
    }

    //根据exchangeName 获取全部绑定对象
    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    //新增绑定对象
    public void insertBinding(Binding binding) throws MqException {
        //先使用exchangeName 对应的哈希表是否存在 不存在就创建一个
//        ConcurrentHashMap<String,Binding> bindingMap=bindingsMap.get(binding.getExchangeName());
//        if(bindingMap==null){
//            //说明不存在 需要手动窗口
//            bindingMap= new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(),bindingMap);
//        }
        //等同替代 拿着key去查找 如果不存在 就创建新的ConcurrentHashMap
        ConcurrentHashMap<String,Binding> bindingMap=bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k->new ConcurrentHashMap<>());
        //因为这里是两步操作 进行移动 所以会有线程安全问题 单独拿出一步都是线程安全的
        synchronized (bindingMap) {
            //进行进一步的插入 再根据queueName 如果已经存在就抛出异常 如果不存在就插入数据
            if (bindingMap.get(binding.getQueueName()) != null) {
                //已经存在这个绑定了 直接抛出异常
                throw new MqException("[MemoryDataCenter] 绑定已经存在 exchangeName=" + binding.getExchangeName() + ", " +
                        "queueName= " + binding.getQueueName());
            }
            //这里的bindingMap 实际上就是bindingsMap中的对象 指定进行修改了
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 绑定添加成功 queueName="+binding.getQueueName()+
                " exchangeName="+binding.getExchangeName());
    }

    //删除绑定对象
    public void deleteBinding(Binding binding) throws MqException {
        //先查询第一层的绑定关系
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap==null){
            //没有查询到 说明没有绑定关系
            throw new MqException("[MemoryDataCenter] 绑定不存在 exchangeName=" + binding.getExchangeName() + ", " +
                    "queueName= " + binding.getQueueName());
        }
        //删除对应的绑定关系
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功 queueName="+binding.getQueueName()+
                " exchangeName="+binding.getExchangeName());
    }

    //添加消息
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新消息添加成功 messageId= "+message.getMessageId());
    }

    //根据id 查询消息
    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    //根据 id 删除消息
    public void removeMessage(String messageId){
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除成功 messageId= "+messageId);
    }

    //发送消息到指定队列
    public void sendMessage(MSGQueue queue,Message message){
        //把消息放到对应的队列数据结构中
        //先根据队列名字 找对队列对应的消息链表
//        LinkedList<Message> messages = queueMessageMap.get(queue.getName());
//        if(message==null){
//            //如果该队列没有数据 需要新增一个数据
//            messages=new LinkedList<>();
//            queueMessageMap.put(queue.getName(),messages);
//        }
        //这个方法和上面注释掉的内容等同效果
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),k->new LinkedList<>());
        synchronized (messages){
            messages.add(message);
            //在这里我们把该消息往消息中心中插入 假设message已经存在了 重复插入也没关系
        }
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息投递到队列中! messageId="+message.getMessageId());
    }

    //从队列中取消息
    public Message pollMessage(String queueName){
        //根据队列名字 查找队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages==null){
            return null;
        }
        synchronized (messages) {
            if (messages.size() == 0) {
                //队列中没有消息
                return null;
            }
            //链表中有元素 就进行头删
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从对列中取出! messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    //获取指定队列中消息的个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages==null){
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

    //添加未确认消息
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String, Message> messageHashMap =
                queueMessageWaitAckMap.computeIfAbsent(queueName,k->new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列 messageId="+message.getMessageId());
    }

    //删除未确认消息
    public void removeMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String, Message> messageHashMap =
                queueMessageWaitAckMap.get(queueName);
        //如果没有这个消息 直接返回
        if(messageHashMap==null){
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除! messageId="+messageId);
    }

    //获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String, Message> messageHashMap =
                queueMessageWaitAckMap.get(queueName);
        //如果没有这个消息 直接返回
        if(messageHashMap==null){
            return null;
        }
        return messageHashMap.get(messageId);
    }

    //从硬盘中取数据 把硬盘中存储的各个维度的数据都恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        //1.恢复所有交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchange();
//        for (int i = 0; i <exchanges.size(); i++) {
//            //将查询结果添加到内存
//            exchangeMap.put(exchanges.get(i).getName(),exchanges.get(i));
//        }
        for(Exchange exchange :exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }
        //2.恢复所有队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueue();
        for(MSGQueue queue :queues){
            queueMap.put(queue.getName(),queue);
        }
        //3.恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBinding();
        for(Binding binding :bindings){
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName()
            ,k->new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }
        //4.恢复所有的消息数据
        //遍历所有队列 根据每个队列的名字 获取所有的消息
        for (MSGQueue queue:queues){
            //读取这个name的队列的所有消息
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            //把每个消息 添加到消息中心
            for(Message message:messages){
                messageMap.put(message.getMessageId(),message);
            }
        }
        //针对未确认的消息 这部分内存中的数据 不需要存在硬盘中
        //此时也就无需恢复 一旦在等待ACK的时候 服务器重启了 我们这些未被确认的消息 就恢复成未被取走的消息了
        //这个消息在硬盘上存储的时候 就是当作 "未被取走"
    }
}
